use async_trait::async_trait;
use rustcloud_core::{
    ServiceRegistry, ServiceInstance, ServiceResult, ServiceError, RegistryType, 
    ServiceWatcher, HealthCheck, ServiceFilter, HealthCheckType
};
use crate::RegistryConfig;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::time::{Duration, SystemTime};
use reqwest::Client;
use tracing::{error, warn, info, debug};
use tokio::time::interval;

// Structure for parsing instance list from Nacos v1 API (standard format)
#[derive(Deserialize, Debug)]
#[serde(rename_all = "camelCase")]
struct NacosDiscoveryResult {
    hosts: Vec<NacosDiscoveredInstance>,
    dom: String,
    name: String,
    cache_millis: u64,
    last_ref_time: u64,
    checksum: String,
    all_i_ps: bool,
    reach_protection_threshold: bool,
    valid: bool,
}

#[derive(Deserialize, Debug)]
#[serde(rename_all = "camelCase")]
struct NacosDiscoveredInstance {
    instance_id: Option<String>,
    ip: String,
    port: u16,
    weight: f64,
    healthy: bool,
    enabled: bool,
    ephemeral: Option<bool>,
    cluster_name: Option<String>,
    service_name: Option<String>,
    metadata: HashMap<String, String>,
}

#[derive(Serialize, Debug)]
#[serde(rename_all = "camelCase")]
struct BeatInfo {
    ip: String,
    port: u16,
    weight: f64,
    service_name: String,
    cluster: String,
    metadata: HashMap<String, String>,
    scheduled: bool,
}


/// Nacos服务注册中心实现
#[derive(Clone)]
pub struct NacosRegistry {
    client: Client,
    config: RegistryConfig,
    heartbeat_tasks: std::sync::Arc<tokio::sync::RwLock<HashMap<String, tokio::task::JoinHandle<()>>>>,
    health_check_tasks: std::sync::Arc<tokio::sync::RwLock<HashMap<String, tokio::task::JoinHandle<()>>>>,
    service_cache: std::sync::Arc<tokio::sync::RwLock<HashMap<String, (Vec<ServiceInstance>, SystemTime)>>>,
    cache_ttl: Duration,
}



impl NacosRegistry {
    pub fn new(config: RegistryConfig) -> ServiceResult<Self> {
        let client = Client::builder()
            .timeout(Duration::from_secs(10))
            .build()
            .map_err(|e| ServiceError::NetworkError(e.to_string()))?;

        if config.registry_type != RegistryType::Nacos {
            return Err(ServiceError::ConfigurationError(
                format!("Expected Nacos registry type, got: {}", config.registry_type)
            ));
        }

        Ok(Self {
            client,
            config,
            heartbeat_tasks: std::sync::Arc::new(tokio::sync::RwLock::new(HashMap::new())),
            health_check_tasks: std::sync::Arc::new(tokio::sync::RwLock::new(HashMap::new())),
            service_cache: std::sync::Arc::new(tokio::sync::RwLock::new(HashMap::new())),
            cache_ttl: Duration::from_secs(30), // 30秒缓存TTL
        })
    }

    fn build_url(&self, endpoint: &str) -> String {
        format!("http://{}/nacos/v1/ns/{}", self.config.server_addr, endpoint)
    }

    async fn send_heartbeat(self, instance: ServiceInstance) {
        let url = self.build_url("instance/beat");
        let service_name = instance.service_id.clone();
        let group_name = self.config.group.clone().unwrap_or_else(|| "DEFAULT_GROUP".to_string());
        let namespace_id = self.config.namespace.clone().unwrap_or_else(|| "public".to_string());
        let cluster_name = self.config.group.clone().unwrap_or_else(|| "DEFAULT".to_string());

        loop {
            tokio::time::sleep(Duration::from_secs(5)).await;

            let beat_info = BeatInfo {
                ip: instance.host.clone(),
                port: instance.port,
                weight: instance.weight,
                service_name: format!("{}@@{}", group_name, service_name),
                cluster: cluster_name.clone(),
                metadata: instance.metadata.clone(),
                scheduled: true,
            };

            let beat_json = match serde_json::to_string(&beat_info) {
                Ok(json) => json,
                Err(e) => {
                    error!("Failed to serialize beat info: {}", e);
                    continue; // Skip this heartbeat attempt
                }
            };

            let params = [
                ("serviceName", service_name.clone()),
                ("groupName", group_name.clone()),
                ("namespaceId", namespace_id.clone()),
                ("ip", instance.host.clone()),
                ("port", instance.port.to_string()),
                ("beat", beat_json),
            ];

            match self.client.put(&url).query(&params).send().await {
                Ok(response) if !response.status().is_success() => {
                    let status = response.status();
                    let text = response.text().await.unwrap_or_default();
                    warn!(
                        "Heartbeat failed for service: {}. Status: {}. Body: {}",
                        service_name,
                        status,
                        text
                    );
                }
                Err(e) => {
                    error!("Heartbeat error for service {}: {}", service_name, e);
                }
                _ => {}
            }
        }
    }

    fn start_heartbeat(&self, instance: ServiceInstance) {
        let heartbeat_tasks = self.heartbeat_tasks.clone();
        let registry = self.clone();
        let instance_id = format!("{}:{}:{}", instance.service_id, instance.host, instance.port);

        tokio::spawn(async move {
            let task = tokio::spawn(registry.send_heartbeat(instance));
            heartbeat_tasks.write().await.insert(instance_id, task);
        });
    }

    async fn stop_heartbeat(&self, instance: &ServiceInstance) {
        let instance_id = format!("{}:{}:{}", instance.service_id, instance.host, instance.port);
        if let Some(task) = self.heartbeat_tasks.write().await.remove(&instance_id) {
            task.abort();
        }
    }

    pub async fn shutdown(&self) {
        // 停止所有心跳任务
        let mut tasks = self.heartbeat_tasks.write().await;
        for (_, task) in tasks.drain() {
            task.abort();
        }
        
        // 停止所有健康检查任务
        let mut health_tasks = self.health_check_tasks.write().await;
        for (_, task) in health_tasks.drain() {
            task.abort();
        }
    }
    
    /// 启动健康检查任务
    async fn start_health_check(&self, instance: ServiceInstance, health_check: HealthCheck) {
        let client = self.client.clone();
        let instance_id = format!("{}:{}:{}", instance.service_id, instance.host, instance.port);
        let health_check_tasks = self.health_check_tasks.clone();
        let registry = self.clone();
        
        tokio::spawn(async move {
            let task = tokio::spawn(async move {
                Self::health_check_loop(client, instance, health_check, registry).await;
            });
            health_check_tasks.write().await.insert(instance_id, task);
        });
    }
    
    /// 健康检查循环
    async fn health_check_loop(
        client: Client, 
        instance: ServiceInstance, 
        health_check: HealthCheck,
        registry: NacosRegistry,
    ) {
        let mut interval = interval(health_check.interval);
        let mut failure_count = 0u32;
        let mut success_count = 0u32;
        
        loop {
            interval.tick().await;
            
            let health_result = match &health_check.check_type {
                HealthCheckType::Http { path, method } => {
                    Self::http_health_check(&client, &instance, path, method, health_check.timeout).await
                }
                HealthCheckType::Tcp => {
                    Self::tcp_health_check(&instance, health_check.timeout).await
                }
                HealthCheckType::Script { command } => {
                    Self::script_health_check(command, health_check.timeout).await
                }
            };
            
            match health_result {
                Ok(is_healthy) => {
                    if is_healthy {
                        success_count += 1;
                        failure_count = 0;
                        
                        // 如果达到成功阈值，更新实例为健康
                        if success_count >= health_check.success_threshold {
                            if let Err(e) = registry.update_instance_health(&instance, true).await {
                                error!("Failed to update instance health to healthy: {}", e);
                            }
                        }
                    } else {
                        failure_count += 1;
                        success_count = 0;
                        
                        // 如果达到失败阈值，更新实例为不健康
                        if failure_count >= health_check.failure_threshold {
                            if let Err(e) = registry.update_instance_health(&instance, false).await {
                                error!("Failed to update instance health to unhealthy: {}", e);
                            }
                        }
                    }
                }
                Err(e) => {
                    error!("Health check failed for instance {}: {}", instance, e);
                    failure_count += 1;
                    success_count = 0;
                }
            }
        }
    }
    
    /// HTTP健康检查
    async fn http_health_check(
        client: &Client,
        instance: &ServiceInstance,
        path: &str,
        method: &str,
        timeout: Duration,
    ) -> ServiceResult<bool> {
        let url = format!("{}://{}{}", instance.scheme, instance.address(), path);
        
        let request = match method.to_uppercase().as_str() {
            "GET" => client.get(&url),
            "POST" => client.post(&url),
            "PUT" => client.put(&url),
            "HEAD" => client.head(&url),
            _ => return Err(ServiceError::ConfigurationError(format!("Unsupported HTTP method: {}", method))),
        };
        
        match tokio::time::timeout(timeout, request.send()).await {
            Ok(Ok(response)) => {
                let is_healthy = response.status().is_success() || response.status().as_u16() == 404;
                Ok(is_healthy)
            }
            Ok(Err(e)) => {
                debug!("HTTP health check failed for {}: {}", url, e);
                Ok(false)
            }
            Err(_) => {
                debug!("HTTP health check timeout for {}", url);
                Ok(false)
            }
        }
    }
    
    /// TCP健康检查
    async fn tcp_health_check(instance: &ServiceInstance, timeout: Duration) -> ServiceResult<bool> {
        use tokio::net::TcpStream;
        
        let addr = format!("{}:{}", instance.host, instance.port);
        
        match tokio::time::timeout(timeout, TcpStream::connect(&addr)).await {
            Ok(Ok(_)) => Ok(true),
            Ok(Err(e)) => {
                debug!("TCP health check failed for {}: {}", addr, e);
                Ok(false)
            }
            Err(_) => {
                debug!("TCP health check timeout for {}", addr);
                Ok(false)
            }
        }
    }
    
    /// 脚本健康检查
    async fn script_health_check(command: &str, timeout: Duration) -> ServiceResult<bool> {
        use tokio::process::Command;
        
        let mut cmd = if cfg!(target_os = "windows") {
            let mut cmd = Command::new("cmd");
            cmd.args(["/C", command]);
            cmd
        } else {
            let mut cmd = Command::new("sh");
            cmd.args(["-c", command]);
            cmd
        };
        
        match tokio::time::timeout(timeout, cmd.output()).await {
            Ok(Ok(output)) => Ok(output.status.success()),
            Ok(Err(e)) => {
                debug!("Script health check failed: {}", e);
                Ok(false)
            }
            Err(_) => {
                debug!("Script health check timeout");
                Ok(false)
            }
        }
    }
    
    /// 更新实例健康状态
    async fn update_instance_health(&self, instance: &ServiceInstance, healthy: bool) -> ServiceResult<()> {
        let url = self.build_url("instance");
        let namespace_id = self.config.namespace.clone().unwrap_or_else(|| "public".to_string());
        let group_name = self.config.group.clone().unwrap_or_else(|| "DEFAULT_GROUP".to_string());
        
        let params = [
            ("serviceName", instance.service_id.clone()),
            ("ip", instance.host.clone()),
            ("port", instance.port.to_string()),
            ("healthy", healthy.to_string()),
            ("namespaceId", namespace_id),
            ("groupName", group_name),
        ];
        
        let response = self.client.put(&url).query(&params).send().await
            .map_err(|e| ServiceError::NetworkError(e.to_string()))?;
            
        if !response.status().is_success() {
            let error_text = response.text().await.unwrap_or_default();
            return Err(ServiceError::RegistryError(format!("Failed to update instance health: {}", error_text)));
        }
        
        info!("Updated instance health: {} -> {}", instance, healthy);
        Ok(())
    }
    
    /// 获取服务列表
    async fn get_services(&self) -> ServiceResult<Vec<String>> {
        let url = self.build_url("service/list");
        let namespace_id = self.config.namespace.clone().unwrap_or_else(|| "public".to_string());
        let group_name = self.config.group.clone().unwrap_or_else(|| "DEFAULT_GROUP".to_string());
        
        let params = [
            ("pageNo", "1".to_string()),
            ("pageSize", "1000".to_string()),
            ("namespaceId", namespace_id),
            ("groupName", group_name),
        ];
        
        let response = self.client.get(&url).query(&params).send().await
            .map_err(|e| ServiceError::NetworkError(e.to_string()))?;
            
        if response.status().is_success() {
            let body: serde_json::Value = response.json().await
                .map_err(|e| ServiceError::SerializationError(e.to_string()))?;
                
            if let Some(doms) = body.get("doms").and_then(|v| v.as_array()) {
                let services: Vec<String> = doms.iter()
                    .filter_map(|v| v.as_str())
                    .map(|s| s.to_string())
                    .collect();
                Ok(services)
            } else {
                Ok(Vec::new())
            }
        } else {
            let error_text = response.text().await.unwrap_or_default();
            Err(ServiceError::RegistryError(format!("Failed to get services: {}", error_text)))
        }
    }
    
    /// 检查服务缓存
    async fn get_cached_instances(&self, service_name: &str) -> Option<Vec<ServiceInstance>> {
        let cache = self.service_cache.read().await;
        if let Some((instances, cached_time)) = cache.get(service_name) {
            if cached_time.elapsed().unwrap_or(Duration::from_secs(u64::MAX)) < self.cache_ttl {
                return Some(instances.clone());
            }
        }
        None
    }
    
    /// 更新服务缓存
    async fn update_cache(&self, service_name: &str, instances: Vec<ServiceInstance>) {
        let mut cache = self.service_cache.write().await;
        cache.insert(service_name.to_string(), (instances, SystemTime::now()));
    }
}

#[async_trait]
impl ServiceRegistry for NacosRegistry {
    async fn register(&self, instance: ServiceInstance) -> ServiceResult<()> {
        let url = format!("http://{}/nacos/v1/ns/instance", self.config.server_addr);

        // 使用实例的分组和命名空间，如果没有则使用配置的默认值
        let namespace_id = instance.namespace
            .clone()
            .or_else(|| self.config.namespace.clone())
            .unwrap_or_else(|| "public".to_string());
            
        let group_name = instance.group
            .clone()
            .or_else(|| self.config.group.clone())
            .unwrap_or_else(|| "DEFAULT_GROUP".to_string());
            
        let cluster_name = instance.cluster
            .clone()
            .unwrap_or_else(|| "DEFAULT".to_string());
        
        // 克隆用于参数的值
        let service_id = instance.service_id.clone();
        let service_id_clone = service_id.clone();
        let host = instance.host.clone();
        let port = instance.port;
        let weight = instance.weight;
        let healthy = instance.healthy;
        
        // 将元数据和环境信息合并
        let mut metadata = instance.metadata.clone();
        
        // 添加环境信息到元数据
        if let Some(environment) = &instance.environment {
            metadata.insert("environment".to_string(), environment.clone());
        }
        
        if let Some(cluster) = &instance.cluster {
            metadata.insert("cluster".to_string(), cluster.clone());
        }
        
        // 将元数据序列化为JSON字符串
        let metadata_json = if metadata.is_empty() {
            "{}".to_string()
        } else {
            serde_json::to_string(&metadata)
                .map_err(|e| ServiceError::SerializationError(e.to_string()))?
        };

        let params = [
            ("serviceName", service_id),
            ("ip", host),
            ("port", port.to_string()),
            ("healthy", healthy.to_string()),
            ("weight", weight.to_string()),
            ("namespaceId", namespace_id.clone()),
            ("namespace", namespace_id.clone()),
            ("groupName", group_name.clone()),
            ("clusterName", cluster_name),
            ("metadata", metadata_json),
        ];

        let response = self.client.post(&url)
            .query(&params)
            .send()
            .await
            .map_err(|e| ServiceError::RegistryError(format!("Failed to register service: {}", e)))?;

        if response.status().is_success() {
            // Start heartbeat for the registered instance
            self.start_heartbeat(instance);
            info!("Successfully registered service instance: {} in group: {}, namespace: {}", 
                  service_id_clone, group_name.clone(), namespace_id.clone());
            Ok(())
        } else {
            let error_text = response.text().await.unwrap_or_else(|_| "Unknown error".to_string());
            Err(ServiceError::RegistryError(format!("Failed to register service: {}", error_text)))
        }
    }
    
    async fn register_with_metadata(&self, mut instance: ServiceInstance, metadata: HashMap<String, String>) -> ServiceResult<()> {
        // 合并元数据
        instance.metadata.extend(metadata);
        self.register(instance).await
    }
    
    async fn register_with_health_check(&self, instance: ServiceInstance, health_check: HealthCheck) -> ServiceResult<()> {
        // 先注册服务实例
        self.register(instance.clone()).await?;
        
        // 然后启动健康检查
        self.start_health_check(instance, health_check).await;
        
        Ok(())
    }

    async fn deregister(&self, instance: ServiceInstance) -> ServiceResult<()> {
        let url = self.build_url("instance");
        let params = [
            ("serviceName", instance.service_id.clone()),
            ("ip", instance.host.clone()),
            ("port", instance.port.to_string()),
            ("namespaceId", self.config.namespace.clone().unwrap_or_else(|| "public".to_string())),
            ("ephemeral", "true".to_string()),
        ];

        let response = self.client.delete(&url).query(&params).send().await
            .map_err(|e| ServiceError::NetworkError(e.to_string()))?;

        if response.status().is_success() {
            self.stop_heartbeat(&instance).await;
            info!("Successfully deregistered service instance: {}", instance);
            Ok(())
        } else {
            let error_text = response.text().await.unwrap_or_default();
            error!("Failed to deregister service: {}", error_text);
            Err(ServiceError::RegistryError(format!("Failed to deregister service: {}", error_text)))
        }
    }

    async fn discover(&self, service_name: &str) -> ServiceResult<Vec<ServiceInstance>> {
        // 检查缓存
        if let Some(cached_instances) = self.get_cached_instances(service_name).await {
            debug!("Returning cached instances for service: {}", service_name);
            return Ok(cached_instances);
        }
        
        let url = self.build_url("instance/list");
        let namespace_id = self.config.namespace.clone().unwrap_or_else(|| "public".to_string());
        let group_name = self.config.group.clone().unwrap_or_else(|| "DEFAULT_GROUP".to_string());
        let clusters = self.config.group.clone().unwrap_or_else(|| "DEFAULT".to_string());
        
        let params = [
            ("serviceName", service_name.to_string()),
            ("groupName", group_name.clone()),
            ("namespaceId", namespace_id.clone()),
            ("namespace", namespace_id.clone()), // Some Nacos versions require this as well
            ("clusters", clusters),
            ("healthyOnly", "false".to_string()), // 获取所有实例，包括不健康的
        ];

        let response = self.client.get(&url).query(&params).send().await
            .map_err(|e| ServiceError::NetworkError(e.to_string()))?;

        if response.status().is_success() {
            let response_text = response.text().await.map_err(|e| ServiceError::NetworkError(e.to_string()))?;
            
            let body: NacosDiscoveryResult = serde_json::from_str(&response_text)
                .map_err(|e| {
                    error!("Failed to deserialize Nacos response. Raw response: '{}'. Error: {}", response_text, e);
                    ServiceError::SerializationError(format!("error decoding response body: {}", e))
                })?;
            
            let instances: Vec<ServiceInstance> = body.hosts.into_iter()
                .filter(|h| h.enabled)
                .map(|h| {
                    let service_name = h.service_name.clone().unwrap_or_else(|| service_name.to_string());
                    let mut instance = ServiceInstance::new(service_name, h.ip.clone(), h.port, "http");
                    instance.weight = h.weight;
                    instance.healthy = h.healthy;
                    instance.metadata = h.metadata.clone();
                    
                    // 从元数据中提取环境信息
                    if let Some(environment) = h.metadata.get("environment") {
                        instance.environment = Some(environment.clone());
                    }
                    
                    if let Some(cluster) = h.metadata.get("cluster") {
                        instance.cluster = Some(cluster.clone());
                    } else {
                        // 使用集群名作为默认值
                        if let Some(cluster_name) = h.cluster_name {
                            instance.cluster = Some(cluster_name);
                        }
                    }
                    
                    // 设置分组和命名空间（从请求参数中获取）
                    instance.namespace = Some(namespace_id.clone());
                    instance.group = Some(group_name.clone());
                    
                    instance
                }).collect();
                
            // 更新缓存
            self.update_cache(service_name, instances.clone()).await;
            
            debug!("Discovered {} instances for service: {}", instances.len(), service_name);
            Ok(instances)
        } else {
            let error_text = response.text().await.unwrap_or_default();
            error!("Failed to discover service: {}", error_text);
            Err(ServiceError::RegistryError(format!("Failed to discover service: {}", error_text)))
        }
    }
    
    async fn discover_with_filter(&self, service_name: &str, filter: ServiceFilter) -> ServiceResult<Vec<ServiceInstance>> {
        let instances = self.discover(service_name).await?;
        let original_count = instances.len();
        let filtered_instances: Vec<ServiceInstance> = instances.into_iter().filter(|instance| {
            // 健康状态过滤
            if filter.healthy_only && !instance.healthy {
                return false;
            }
            
            // 元数据过滤
            for (key, value) in &filter.metadata_filters {
                if instance.metadata.get(key) != Some(value) {
                    return false;
                }
            }
            
            // 版本过滤
            if let Some(version) = &filter.version_filter {
                if instance.metadata.get("version") != Some(version) {
                    return false;
                }
            }
            
            // 集群过滤
            if let Some(cluster) = &filter.cluster_filter {
                if instance.metadata.get("cluster") != Some(cluster) {
                    return false;
                }
            }
            
            true
        }).collect();
        
        debug!("Filtered {} instances for service: {} (original: {})", 
               filtered_instances.len(), service_name, original_count);
        Ok(filtered_instances)
    }

    async fn health_check(&self, service_name: &str) -> ServiceResult<Vec<ServiceInstance>> {
        let instances = self.discover(service_name).await?;
        let healthy_instances = instances.into_iter().filter(|instance| instance.healthy).collect();
        Ok(healthy_instances)
    }
    
    async fn list_services(&self) -> ServiceResult<Vec<String>> {
        self.get_services().await
    }
    
    async fn registry_health(&self) -> ServiceResult<bool> {
        // 通过检查 Nacos 服务器的健康状态来判断
        let url = format!("http://{}/nacos/actuator/health", self.config.server_addr);
        
        match self.client.get(&url).timeout(Duration::from_secs(5)).send().await {
            Ok(response) if response.status().is_success() => Ok(true),
            _ => Ok(false), // 任何错误都认为不健康
        }
    }

    async fn watch(&self, _service_name: &str) -> ServiceResult<Box<dyn ServiceWatcher + Send>> {
        // TODO: 实现 Nacos 服务监听功能
        Err(ServiceError::NotImplemented("Watch API not implemented for Nacos yet".to_string()))
    }
}